package com.hbase.tableoperation;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 导出特定列的数据，针对小数据量
 * 
 * @author LBH
 *
 */
public class DiffHbase {
	public static Configuration configuation = null;
	public static Connection hbaseConnection = null;

	static {
		configuation = HBaseConfiguration.create();
		configuation.set("hbase.zookeeper.quorum", "192.168.51.253");
		// configuation.set("hbase.zookeeper.property.clientPort", "2181");

	}

	public static void init() {
		try {
			hbaseConnection = ConnectionFactory.createConnection(configuation);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void selectRowKey(TableName tableName, String rowKey) {

		try {
			Table table = hbaseConnection.getTable(tableName);
			Get get = new Get(rowKey.getBytes());
			Result result = table.get(get);

			for (KeyValue keyValue : result.raw()) {
				System.out.println("------------" + String.valueOf(keyValue.getRow()) + "-----------");
				System.out.println("Column Family:" + String.valueOf(keyValue.getFamily()));
				System.out.println("value:" + String.valueOf(keyValue.getValue()));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void selectRowKeyFamily(TableName tableName, String rowKey, String family) {
		try {
			Table table = hbaseConnection.getTable(tableName);
			Get get = new Get(rowKey.getBytes());
			get.addFamily(family.getBytes());

			Result result = table.get(get);
			for (Cell cell : result.rawCells()) {
				System.out.println("------------" + String.valueOf(CellUtil.cloneRow(cell)) + "-----------");
				System.out.println("Column Family:" + String.valueOf(cell.getFamily()));
				System.out.println("value:" + String.valueOf(CellUtil.cloneValue(cell)));
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void selectRowKeyFamilyColumn(TableName tableName, String rowKey, String family, String column) {
		try {
			Table table = hbaseConnection.getTable(tableName);
			Get get = new Get(rowKey.getBytes());
			get.addColumn(family.getBytes(), column.getBytes());

			Result result = table.get(get);
			for (Cell cell : result.rawCells()) {
				System.out.println("------------" + String.valueOf(cell.getRow()) + "-----------");
				System.out.println("Column Family:" + String.valueOf(cell.getFamily()));
				System.out.println("Column:" + String.valueOf(cell.getQualifier()));
				System.out.println("value:" + String.valueOf(cell.getValue()));

			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	private static final String USAGE = "Usage:DiffHbase [-o outfile] tablename infile filterColumns...";

	private static void printUsage(String message) {
		System.err.println(message);
		System.err.println(USAGE);
		throw new RuntimeException(USAGE);
	}

	private static void printId(String id, Result result) {
		String value = Bytes
				.toString(result.getValue(ColumnUtils.getFamily("info:url"), ColumnUtils.getQualifier("info:url")));
		if (value == null) {
			System.out.println(id + "\t NULL");
		} else {
			System.out.println(id + "\t " + value);
		}
	}

	private static void writeId(String id, Result result, FileOutputStream fos) {
		String value = Bytes
				.toString(result.getValue(ColumnUtils.getFamily("info:url"), ColumnUtils.getQualifier("info:url")));

		try {
			if (value == null) {
				fos.write((id + "\t NULL").getBytes());
			} else {
				fos.write((id + "\t " + value).getBytes());
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}
	
	private static void printRow(String id,Result result){
		System.out.println("------------" + id + "-----------");

		for (Cell cell : result.rawCells()) {
			System.out.println(String.valueOf(CellUtil.cloneFamily(cell))+":"
		          +String.valueOf(CellUtil.cloneQualifier(cell))
		          +String.valueOf(CellUtil.cloneValue(cell)));
		}
	}

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		if(args.length<3){
			printUsage("too few arguments.");
		}
		String outfile=null;
		TableName tableName=TableName.valueOf(args[0]);
		String dictfile=args[1];
		int skilLen=2;
		
		if(args[0].equals("-o")){
			outfile=args[1];
			tableName=TableName.valueOf(args[2]);
			dictfile=args[3];
			skilLen=4;
		}

		try {
			Table table=hbaseConnection.getTable(tableName);
			String[] filterColumns=new String[args.length-skilLen];
			System.arraycopy(args, skilLen, filterColumns, 0, args.length-skilLen);
			
			System.out.println("filterColumns:");
			for(int i=0;i<filterColumns.length;i++){
				System.out.println("\t"+filterColumns[i]);
			}
			
			FileOutputStream fos=null;
			if(outfile!=null){
				fos=new FileOutputStream(outfile);
			}
			
			int count=0;
			SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			
			File srcFile=new File(dictfile);
			FileInputStream fin=new FileInputStream(srcFile);
			InputStreamReader isr=new InputStreamReader(fin);
			BufferedReader bufferReader=new BufferedReader(isr);
			String read=null;
			
			while((read=bufferReader.readLine())!=null){
				String[] split=read.trim().split("\\s");
				if(split.length<1){
					System.out.println("Error line:"+read);
					continue;
				}
				if(++count%1000==0){
					System.out.println(sdf.format(new Date())+" : "+count+" rows processed");
				}
				
				Get get=new Get(split[0].getBytes());
				Result result=table.get(get);
				if(result==null){
					System.out.println("No result for "+split[0]);
					continue;
				}
				
				for(int i=0;i<filterColumns.length;i++){
					String value=Bytes.toString(result.getValue(
							ColumnUtils.getFamily(filterColumns[i]), 
							ColumnUtils.getQualifier((filterColumns[i]))));
					
					if(value==null){
						if(fos==null){
							printId(split[0],result);
						}else{
							writeId(split[0],result,fos);
						}
						break;
					}
				}
			}
					
			bufferReader.close();
			isr.close();
			fin.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}

}

class ColumnUtils {
	public static byte[] getFamily(String column) {
		return getBytes(column, 0);
	}

	public static byte[] getQualifier(String column) {
		return getBytes(column, 1);
	}

	private static byte[] getBytes(String column, int offset) {
		String[] split = column.split(":");
		return Bytes.toBytes(offset > split.length - 1 ? split[0] : split[offset]);
	}
}
